package com.peng.rpc.util;

import com.alibaba.nacos.common.utils.StringUtils;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.*;

/**
 * 创建线程池的工具类，简化SocketServer中创建线程池的代码
 * @author Wonderson
 * @email wpp.ds@foxmail.com
 * @date 2022-03-03 19:49
 */
@Slf4j
public class ThreadPoolFactory {

    /**
     * 线程池参数
     */
    private static final int CORE_POOL_SIZE = 10;
    private static final int MAXIMUM_POOL_SIZE_SIZE = 100;
    private static final int KEEP_ALIVE_TIME = 1;
    private static final int BLOCKING_QUEUE_CAPACITY = 100;

    private static final Map<String, ExecutorService> threadPoolMap = new ConcurrentHashMap<>();

    public static ExecutorService createDefaultThreadPool(String threadPoolPrefix) {
        return createDefaultThreadPool(threadPoolPrefix, false);
    }

    public static ExecutorService createDefaultThreadPool(String threadPoolPrefix, boolean isDaemon) {

        ExecutorService threadPool = threadPoolMap.computeIfAbsent(threadPoolPrefix,
                k -> {
                    log.info("创建新线程池：{}", threadPoolPrefix);
                    return createThreadPool(threadPoolPrefix, isDaemon);
                });
        if (threadPool.isTerminated() || threadPool.isShutdown()) {
            threadPool = createThreadPool(threadPoolPrefix, isDaemon);
            threadPoolMap.put(threadPoolPrefix, threadPool);
        }
        return threadPool;
    }

    public static ExecutorService createThreadPool(String threadPoolPrefix, boolean isDaemon) {
        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(BLOCKING_QUEUE_CAPACITY);
        ThreadFactory threadPoolFactory = createThreadPoolFactory(threadPoolPrefix, isDaemon);
        return new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE_SIZE, KEEP_ALIVE_TIME,
                TimeUnit.MINUTES, blockingQueue, threadPoolFactory);
    }

    private static ThreadFactory createThreadPoolFactory(String threadPoolPrefix, boolean isDaemon) {
        if (StringUtils.isNotEmpty(threadPoolPrefix)) {
            return new ThreadFactoryBuilder().setDaemon(isDaemon)
                    .setNameFormat(threadPoolPrefix + "-%d").build();
        } else {
            return Executors.defaultThreadFactory();
        }
    }

    /**
     * 关闭所有的线程池
     */
    public static void shutdownAll() {
        log.info("开始关闭所有的线程池。。。");
        threadPoolMap.keySet().parallelStream().forEach(threadPoolName -> {
            ExecutorService executorService = threadPoolMap.get(threadPoolName);
            executorService.shutdown();
            log.info("关闭线程池：{}{}", threadPoolName, executorService.isTerminated());
            try {
                executorService.awaitTermination(10, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                log.info("关闭线程池：{}失败，开始强制关闭", threadPoolName);
                executorService.shutdownNow();
            }
        });
    }

}
